package com.o2o.cleaning.month.platform.ebusiness_plat.jumei_2019_7.save_to_es

import com.alibaba.fastjson.JSON
import org.apache.spark.sql.SparkSession
import org.elasticsearch.spark._

object Jumei_to_157 {
  def main(args: Array[String]): Unit = {

    val spark = SparkSession.builder()
      .appName("Jumei_to_157")
      .master("local[*]")
//      .config("es.nodes", "192.168.1.157")
      .config("es.nodes", "192.168.1.29")
      .config("es.port", "9200")
      .config("cluster.name","O2OElastic")
      .config("es.net.http.auth.user", "elastic")
      .config("es.net.http.auth.pass", "changeme")
      .getOrCreate()
    val sc = spark.sparkContext
    sc.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    sc.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    sc.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    sc.setLogLevel("WARN")
    val platformName = "jumei"
    val year = 2021
    val month = 6
    // 正常跑 ------路径
    val platform_goodUrl= s"s3a://o2o-dataproces-group/zyf/${year}/${month}/${platformName}_result/"
    // 618 ------路径
//    val platform_goodUrl= s"s3a://o2o-dataproces-group/zyf/2021/620/jumei_2021_6/"
      val data = spark.read.orc(platform_goodUrl)
        .where("priceText > 0 and salesAmount > 0 and sellCount > 0")

    val value = data.toJSON.rdd.map(line => {
          val lines=JSON.parseObject(line)
          val function = lines.getOrDefault("function","").toString

          lines.put("function",function)
          lines.put("Base_Info",lines.get("Base_Info").toString)
          lines.put("sellCount",lines.get("sellCount").toString.toDouble.toInt)

          lines.remove("_id")
          lines.remove("start_sell")
          lines.remove("add_to_field")
          lines.remove("end_sell")
          lines.remove("goodRatePercentage")
          lines
        })

    try {
//      value.saveToEs("jumei_2021_activity/jumei_2021_618",
        //    value.saveToEs("jumei_2020_activity/jumei_2020_0615",
            value.saveToEs(s"${year}_jumei/jumei_${year}_${month}",
        Map("es.mapping.id" -> "good_id", "es.nodes" -> s"192.168.1.29",
          "es.port" -> "9200",
          "cluster.name" -> "Es-OTO-Data"))
    }catch {
      case ex: Exception =>{
        println("Exception"+ex.toString)
      }
    }
//4653202
      sc.stop()
  }
}